/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.trp.report.merge;

import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;

import com.semidata.trp.report.data.ReportData;
import com.semidata.trp.report.data.UserData;
import com.semidata.trp.report.sort.DataProvider;
import com.semidata.trp.report.sort.FileDataProvider;
import com.semidata.trp.report.sort.MergeSort;
import com.semidata.trp.report.utils.SqlUtils;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class UserDataMerge {

	private char[] b64 = {'+', '-', '0', '1', '2', '3', '4', '5', '6', '7',
	        '8', '9', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K',
	        'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X',
	        'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k',
	        'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x',
	        'y', 'z'};

	public boolean merge(int threadNum, final List<File> srcDirs,
	        final File destDir, final boolean complete,
	        final ReportData reportData,
	        final Map<String, Map<Integer, Integer>> businessMap) {

		final BlockingQueue<Character> queue = new LinkedBlockingQueue<Character>();
		final ReentrantLock lock = new ReentrantLock();

		for (int i = 0; i < 64; i++)
			queue.add(b64[i]);

		Thread[] threads = new Thread[threadNum];
		
		for (int i = 0; i < threads.length; i++) {

			threads[i] = new Thread(new Runnable() {

				@Override
				public void run() {

					ReportData subReportData = null;
					if (reportData != null)
						subReportData = new ReportData();

					while (true) {
						Character suffix = queue.poll();
						if (suffix == null)
							break;

						MergeOutput out = null;
						if (destDir != null) {
							out = new MergeOutput(destDir, suffix);
						}

						List<DataProvider<String>> providers = new ArrayList<>();

						for (File dir : srcDirs) {

							File file = new File(dir, "data_" + suffix);
							if (file.exists())
								providers.add(new FileDataProvider(file));
						}

						MergeSort<String> sorter = new MergeSort<String>(
						        providers, new Comparator<String>() {

							        @Override
							        public int compare(String s1, String s2) {
							        	String key1 = s1.substring(0, s1.indexOf('\t'));
							        	String key2 = s2.substring(0, s2.indexOf('\t'));
								        return key1.compareTo(key2);
							        }
						        });

						UserData currentData = null;

						String line;
						while ((line = sorter.poll()) != null) {

							UserData userData = new UserData();
							userData.load(line);

							if (currentData == null) {
								currentData = userData;
							} else if (currentData.getUserId().equals(
							        userData.getUserId())) {
								currentData.merge(userData);
							} else {
								if (out != null) {
									out.write(currentData.toString());
								}
								if (subReportData != null) {
									subReportData.merge(currentData,
									        businessMap);
								}
								currentData = userData;
							}
						}

						if (currentData != null) {
							if (out != null) {
								out.write(currentData.toString());
								out.close();
							}

							if (subReportData != null) {
								subReportData.merge(currentData, businessMap);
							}
						}

					}

					lock.lock();
					try {
						if (reportData != null)
							reportData.merge(subReportData);
					} finally {
						lock.unlock();
					}
				}
			});
			threads[i].start();
		}

		for (int i = 0; i < threads.length; i++) {
			try {
				threads[i].join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return complete;
	}

	public static void main(String[] args) {

		UserDataMerge merge = new UserDataMerge();

		File dir = new File(
		        "/Users/shewei/Servers/d189/report/minute/201407/20140726/2014072600");

		File[] srcDirs = dir.listFiles();

		for (File file : srcDirs) {
			System.out.println(file.getName());
		}

		File destDir = new File("/Users/shewei/Servers/d189/report/tmp");
		destDir.mkdirs();

		merge.merge(2, Arrays.asList(srcDirs), destDir, true, null,
		        SqlUtils.getBusinessMap());

	}
}
